package com.rds.benchmark.thread;

import com.rds.benchmark.command.CommandOperator;
import com.rds.benchmark.config.ArgvsConfig;
import com.rds.benchmark.connector.Connector;
import com.rds.benchmark.operator.RedisBenchmarkOperator;
import com.rds.benchmark.operator.TypeOperator;
import io.lettuce.core.api.StatefulConnection;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import org.redisson.api.RedissonClient;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.util.JedisClusterCRC16;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;

public class MainRunThread extends Thread {

    private TypeOperator[] operators;
    private String type;
    private CountDownLatch latch;
    private int final_avg_req;
    private int[] thread_req;

    public MainRunThread(String name) {
        super(name);
    }

    public void initParameter(TypeOperator[] operators, String type, CountDownLatch latch, int final_avg_req, int[] thread_req) {
        this.operators = operators;
        this.type = type;
        this.latch = latch;
        this.final_avg_req = final_avg_req;
        this.thread_req = thread_req;
    }

    @Override
    public void run() {
        int idx = Integer.parseInt(Thread.currentThread().getName());
        int pxx_idx = idx % CommandOperator.PXX_MAX;
        String prefix = UUID.randomUUID().toString();
        String key = CommandOperator.createKey(prefix);
        TypeOperator operator = operators[idx];
        Object conn = null;
        try {
            conn = Connector.connector.getJedis();

            if (conn instanceof JedisCluster) {
                key = key + "{" + idx + "}";
                conn = new Jedis(((JedisCluster) conn).getConnectionFromSlot(JedisClusterCRC16.getSlot(key)));
            } else if (conn instanceof RedissonClient) {
                conn = ((RedissonClient) conn).getBucket(key);
            } else if (conn instanceof StatefulConnection) {
                if (conn instanceof StatefulRedisConnection)
                    conn = ((StatefulRedisConnection) conn).sync();
                else if (conn instanceof StatefulRedisClusterConnection)
                    conn = ((StatefulRedisClusterConnection) conn).sync();
            }

            if (ArgvsConfig.requests == null || !ArgvsConfig.random)
                operator.init(conn, key, idx, type);

            //System.out.println("Thread-" + idx + " barriers waiting count:" + startBarrier.getNumberWaiting());
            Connector.startBarrier.await();

            //操作前清理之前的数据
            // operator.clear();

            // 这里是按照持续时间作为判断条件
            // 在这里添加基于requests的条件
            /**
             * 首先requests和milliTime两个值只能有一个生效，可以默认让milliTime生效。
             * requests生效，按照当前指定的线程分一下平均每个线程要发送多少个请求。
             * 1、当requests数量小于max_threads的数量时候，max_threads的数量等于requests
             * 2、让requests \ max_threads 计算出来的值作为循环条件替换这里即可
             */
            if (ArgvsConfig.requests == null) {

                long stop = System.currentTimeMillis() + ArgvsConfig.milliTime;
                while (stop > System.currentTimeMillis()) {
                    long sub_start = System.nanoTime();

                    operator.operate();

                    sub_start = (System.nanoTime() - sub_start) / 1000;
                    if (sub_start > 0) {
                        CommandOperator.PXXES[pxx_idx].set(sub_start);
                    }
                }
            } else {
                int for_num = final_avg_req;

                if (for_num == 0) {
                    for_num = thread_req[idx];
                }

                for (int j = 0; j < for_num; j++) {
                    long sub_start = System.nanoTime();
                    String newKey = key + idx + j;
                    if (ArgvsConfig.random) {
                        if (ArgvsConfig.cmd == 99) {
                            ((RedisBenchmarkOperator) operator).setNumber(j);
                            if (ArgvsConfig.randomCommands.contains(type)) {
                                ((RedisBenchmarkOperator) operator).setFlag(true);
                                operator.init(conn, key, idx, type);
                            }else {
                                operator.init(conn, newKey, idx, type);
                            }
                        } else {
                            if (type.equals("read")) operator.init(conn, newKey, idx, type);
                            else operator.init(conn, newKey, idx, "req");
                        }
                    }
                    operator.operate();

                    sub_start = (System.nanoTime() - sub_start) / 1000;
                    if (sub_start > 0) {
                        CommandOperator.PXXES[pxx_idx].set(sub_start);
                    }
                }
            }
        } catch (Throwable t) {
            System.err.println("Thread-" + idx + " run failed!");
//                        t.printStackTrace();
        } finally {
            Connector.connector.close(conn);
            if (ArgvsConfig.requests == null)
                operator.clear(); //清除测试数据
            operator.endTimes();
        }
        latch.countDown();
    }
}
